package per.chao.mqtt.domain.protocol.subscribe;

import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.*;
import io.netty.util.AttributeKey;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import per.chao.mqtt.domain.protocol.BaseProtocolHandler;
import per.chao.mqtt.domain.repo.RetainMessageRepo;
import per.chao.mqtt.domain.repo.SubscribeMessageRepo;
import per.chao.mqtt.domain.shared.store.RetainMessageStore;
import per.chao.mqtt.domain.shared.store.SubscribeStore;

import java.util.ArrayList;
import java.util.List;

/**
 * Description:
 *
 * @author W.Chao
 * @date 2021/6/19 23:05
 **/
@Component
@Slf4j
public class Subscribe extends BaseProtocolHandler {
	@Autowired
	private RetainMessageRepo retainMessageRepo;
	@Autowired
	private SubscribeMessageRepo subscribeMessageRepo;

	@Override
	public void handleMessage(Channel channel, Object msg) {
		MqttSubscribeMessage subscribeMessage = (MqttSubscribeMessage) msg;
		List<MqttTopicSubscription> topicSubscriptions = subscribeMessage.payload().topicSubscriptions();
		if (validTopicFilter(topicSubscriptions)) {
			String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
			List<Integer> mqttQoSList = new ArrayList<>();
			for (MqttTopicSubscription topicSubscription : topicSubscriptions) {
				String topic = topicSubscription.topicName();
				MqttQoS mqttQoS = topicSubscription.qualityOfService();
				MqttSubAckMessage mqttSubAckMessage = (MqttSubAckMessage) MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.SUBACK, false, mqttQoS, false, 0),
						MqttMessageIdVariableHeader.from(subscribeMessage.variableHeader().messageId()),
						new MqttSubAckPayload(mqttQoSList)
				);
				channel.writeAndFlush(mqttSubAckMessage);

				sendRetainMessage(channel, topic, mqttQoS);

				SubscribeStore subscribeStore = new SubscribeStore();
				subscribeStore.setClientId(clientId);
				subscribeStore.setMqttQoS(mqttQoS.value());
				subscribeStore.setTopicFilter(topic);
				mqttQoSList.add(mqttQoS.value());
				subscribeMessageRepo.put(topic, clientId, subscribeStore);
				log.info("SUBSCRIBE - clientId: {}, topicFilter: {}, QoS: {}", clientId, topic, mqttQoS.value());
			}
		} else {
			channel.close();
		}
	}

	private void sendRetainMessage(Channel channel, String topic, MqttQoS mqttQoS) {
		List<RetainMessageStore> retainMessageStores = retainMessageRepo.search(topic);
		retainMessageStores.forEach(retainMessageStore -> {
			MqttQoS respQoS = retainMessageStore.getMqttQoS() > mqttQoS.value() ? mqttQoS : MqttQoS.valueOf(retainMessageStore.getMqttQoS());
			if (respQoS == MqttQoS.AT_MOST_ONCE) {
				MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, false, 0),
						new MqttPublishVariableHeader(retainMessageStore.getTopic(), 0), Unpooled.buffer().writeBytes(retainMessageStore.getMessageBytes()));
				log.info("PUBLISH - clientId: {}, topic: {}, Qos: {}", channel.attr(AttributeKey.valueOf("clientId")).get(), retainMessageStore.getTopic(), respQoS.value());
				channel.writeAndFlush(publishMessage);
			}

			if (respQoS == MqttQoS.AT_LEAST_ONCE) {
				int messageId = 0;
				MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, false, 0),
						new MqttPublishVariableHeader(retainMessageStore.getTopic(), messageId), Unpooled.buffer().writeBytes(retainMessageStore.getMessageBytes()));
				log.info("PUBLISH - clientId: {}, topic: {}, Qos: {}, messageId: {}", channel.attr(AttributeKey.valueOf("clientId")).get(), retainMessageStore.getTopic(), respQoS.value(), messageId);
				channel.writeAndFlush(publishMessage);
			}

			if (respQoS == MqttQoS.EXACTLY_ONCE) {
				int messageId = 0;
				MqttPublishMessage publishMessage = (MqttPublishMessage) MqttMessageFactory.newMessage(
						new MqttFixedHeader(MqttMessageType.PUBLISH, false, respQoS, false, 0),
						new MqttPublishVariableHeader(retainMessageStore.getTopic(), messageId), Unpooled.buffer().writeBytes(retainMessageStore.getMessageBytes()));
				log.info("PUBLISH - clientId: {}, topic: {}, Qos: {}, messageId: {}", channel.attr(AttributeKey.valueOf("clientId")).get(), retainMessageStore.getTopic(), respQoS.value(), messageId);
				channel.writeAndFlush(publishMessage);
			}
		});
	}

	private boolean validTopicFilter(List<MqttTopicSubscription> topicSubscriptions) {
		for (MqttTopicSubscription topicSubscription : topicSubscriptions) {
			String topic = topicSubscription.topicName();
			// 以#或+开头的、以/符号结尾的订阅按非法订阅处理，没有按照标准协议实现
			if (topic.startsWith("+")
					|| topic.startsWith("#")
					|| topic.endsWith("/")) {
				return false;
			}

			if (topic.contains("#")) {
				// 出现多个#时按照非法订阅处理
				if (StringUtils.countMatches(topic, "#") > 1) {
					return false;
				}
			}

			if (topic.contains("+")) {
				// 如何+符号和/+字符串出现的次数不等的情况按照非法订阅处理
				if (StringUtils.countMatches(topic, "+") != StringUtils.countMatches(topic, "/+")) {
					return false;
				}
			}
		}

		return true;
	}
}
